
/**
 * @file:       thread_pool.cpp
 * @author:     WubinWang
 * @contact:    wubinstu@163.com
 * @date:       2025-01-01
 * @license:    MIT License
 *
 * Copyright (c) 2025 WubinWang
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 *
 * DISCLAIMER:
 * 1. User acknowledges all application risks and consequences arising from usage
 *    of this software are exclusively borne by the user.
 * 2. The original author shall not be held liable for any direct, indirect,
 *    incidental, special, exemplary or consequential damages.
 * 3. No warranty of fitness for particular purpose or non-infringement is provided. */


#include "thread_pool.hpp"

//--------------------------------------------------------------------------------------------------------------------------//
//------------------------------------------------------Task_Item_st--------------------------------------------------------//
//--------------------------------------------------------------------------------------------------------------------------//

Thread_Pool_ct::Task_Item_st::Task_Item_st (Task_ID_t _task_id, Task_Priority_t _task_priority, Task_Name_t _task_name) :
    _m_task_id {_task_id}, _m_task_priority {_task_priority}, _m_task_name {_task_name} {
    this->_m_ctime = std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();
}

auto Thread_Pool_ct::Task_Item_st::Task_Comparator::operator() (const Task_Item_st & _left, const Task_Item_st & _right) const -> bool {
    return _left._m_task_priority < _right._m_task_priority;
}

auto Thread_Pool_ct::Task_Item_st::id () noexcept -> Task_ID_t & {
    return this->_m_task_id;
}

auto Thread_Pool_ct::Task_Item_st::priority () noexcept -> Task_Priority_t & {
    return this->_m_task_priority;
}

auto Thread_Pool_ct::Task_Item_st::name () noexcept -> Task_Name_t & {
    return this->_m_task_name;
}

auto Thread_Pool_ct::Task_Item_st::entry () noexcept -> Task_Entry_t & {
    return this->_m_entry;
}

auto Thread_Pool_ct::Task_Item_st::callback () noexcept -> Task_Callback_t & {
    return this->_m_callback;
}

auto Thread_Pool_ct::Task_Item_st::time_create () noexcept -> Time_ANY_t & {
    return this->_m_ctime;
}

auto Thread_Pool_ct::Task_Item_st::time_attach () noexcept -> Time_ANY_t & {
    return this->_m_atime;
}

auto Thread_Pool_ct::Task_Item_st::time_process () noexcept -> Time_ANY_t & {
    return this->_m_ptime;
}

auto Thread_Pool_ct::Task_Item_st::time_finish () noexcept -> Time_ANY_t & {
    return this->_m_ftime;
}

auto Thread_Pool_ct::Task_Item_st::time_cpu () noexcept -> Time_ANY_t & {
    return this->_m_cpu_time;
}

auto Thread_Pool_ct::Task_Item_st::time_wall () noexcept -> Time_ANY_t & {
    return this->_m_wall_time;
}

auto Thread_Pool_ct::Task_Item_st::state () noexcept -> Thread_Pool_ct::Task_State_t & {
    return this->_m_state;
}

auto Thread_Pool_ct::Task_Item_st::state_ignorable () noexcept -> bool {
    return this->_m_state == Thread_Pool_ct::Task_State_t::ignorable;
}

auto Thread_Pool_ct::Task_Item_st::state_newlyBuild () noexcept -> bool {
    return this->_m_state == Thread_Pool_ct::Task_State_t::newlyBuild;
}

auto Thread_Pool_ct::Task_Item_st::state_onProcessing () noexcept -> bool {
    return this->_m_state == Thread_Pool_ct::Task_State_t::onProcessing;
}

auto Thread_Pool_ct::Task_Item_st::state_isFinished () noexcept -> bool {
    return this->_m_state == Thread_Pool_ct::Task_State_t::isFinished;
}

auto Thread_Pool_ct::Task_Item_st::thread_id () noexcept -> std::thread::id & {
    return this->_m_process_thread;
}

auto Thread_Pool_ct::Task_Item_st::run () noexcept -> void {
    this->_m_process_thread = std::this_thread::get_id ();  // 某线程调用了 run (), 则该线程 ID 应该被记录

    auto _start             = GetThreadCpuTimeUs ();
    this->_m_ptime          = std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();  // 记录开始处理时间点
    this->_m_state          = Thread_Pool_ct::Task_State_t::onProcessing;                                                                              // 标记任务开始执行

    void * _returned        = nullptr;

    try {
        if (this->_m_entry != nullptr)
            _returned = this->_m_entry ();  // 执行主任务
    } catch (...) {
        this->_m_state = Thread_Pool_ct::Task_State_t::exceptionMain;
        return;
    }

    auto _end          = GetThreadCpuTimeUs ();
    this->_m_ftime     = std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();  // 记录结束处理时间点
    this->_m_state     = Thread_Pool_ct::Task_State_t::isFinished;                                                                                // 标记任务完成
    this->_m_cpu_time  = _end - _start;
    this->_m_wall_time = this->_m_ftime - this->_m_ptime;

    try {
        if (this->_m_callback != nullptr)  // 执行对应回调函数
            this->_m_callback (_returned);
    } catch (...) {
        this->_m_state = Thread_Pool_ct::Task_State_t::exceptionCallback;
    }
}

auto Thread_Pool_ct::Task_Item_st::clear () noexcept -> void {
    this->_m_task_name.clear ();
    this->_m_task_id        = 0;
    this->_m_task_priority  = 0;
    this->_m_entry          = nullptr;
    this->_m_callback       = nullptr;
    this->_m_ctime          = 0;
    this->_m_atime          = 0;
    this->_m_ptime          = 0;
    this->_m_ftime          = 0;
    this->_m_state          = Thread_Pool_ct::Task_State_t::ignorable;
    this->_m_process_thread = std::thread::id {0};
}

auto Thread_Pool_ct::Task_Builder_st::id (Task_ID_t _task_id) noexcept -> Task_Builder_st & {
    this->_task.id () = _task_id;
    return *this;
}

auto Thread_Pool_ct::Task_Builder_st::priority (Task_Priority_t _task_priority) noexcept -> Task_Builder_st & {
    this->_task.priority () = _task_priority;
    return *this;
}

auto Thread_Pool_ct::Task_Builder_st::name (Task_Name_t _task_name) noexcept -> Task_Builder_st & {
    this->_task.name () = _task_name;
    return *this;
}

auto Thread_Pool_ct::Task_Builder_st::entry (Task_Entry_t _entry) noexcept -> Task_Builder_st & {
    this->_task.entry () = _entry;
    return *this;
}

auto Thread_Pool_ct::Task_Builder_st::callback (Task_Callback_t _callback) noexcept -> Task_Builder_st & {
    this->_task.callback () = _callback;
    return *this;
}

auto Thread_Pool_ct::Task_Builder_st::build () const noexcept -> Task_Item_st {
    return this->_task;
}

//--------------------------------------------------------------------------------------------------------------------------//
//------------------------------------------------------Thread_Pool_ct------------------------------------------------------//
//--------------------------------------------------------------------------------------------------------------------------//

Thread_Pool_ct::Thread_Pool_ct (Pool_Config_st _config) :
    _m_thread_max {_config._thread_max}, _m_thread_min {_config._thread_min},
    _m_thread_step {_config._thread_step}, _m_queue_max {_config._queue_max}, _m_auto_threshold {_config._auto_threshold} {
    this->init ();
}

auto Thread_Pool_ct::CountsOF_CoreThreads () -> std::uint32_t {
    return std::thread::hardware_concurrency ();
}

auto Thread_Pool_ct::CountsOF_MinThreads () -> std::uint32_t {
    return std::thread::hardware_concurrency ();
}

auto Thread_Pool_ct::CountsOF_MaxThreads (long double _avg_wait_time, long double _avg_exec_time) -> std::uint32_t {
    const std::uint32_t coreCount = std::thread::hardware_concurrency ();
    if (_avg_exec_time <= 0.0L)
        return coreCount;  // fallback

    long double ratio = _avg_wait_time / _avg_exec_time;

    // 限制波动范围
    if (ratio < 0.0L) ratio = 0.0L;
    if (ratio > 4.0L) ratio = 4.0L;  // 最多 5x core 数

    std::uint32_t maxThreads = static_cast<std::uint32_t> (coreCount * (1.0L + ratio));

    // 保底逻辑
    if (maxThreads < 2) maxThreads = 2;
    return maxThreads;
}

auto Thread_Pool_ct::EMAverage (std::int64_t _accumulated, std::int64_t _new_data, double _alpha) -> std::int64_t {
    if (_alpha < 0.0) _alpha = 0.0;
    if (_alpha > 1.0) _alpha = 1.0;
    if (_accumulated == 0)
        return _new_data;
    double v = _alpha * static_cast<double> (_new_data) + (1.0 - _alpha) * static_cast<double> (_accumulated);
    return static_cast<std::int64_t> (v + 0.5);
}

auto Thread_Pool_ct::time_start () noexcept -> std::int64_t {
    return this->_m_time_start;
}

auto Thread_Pool_ct::time_stop () noexcept -> std::int64_t {
    return this->_m_time_stop;
}

auto Thread_Pool_ct::time_accumulated_running () noexcept -> std::int64_t {
    return this->_m_time_accumulated_running;
}

auto Thread_Pool_ct::time_accumulated_tasking () noexcept -> std::int64_t {
    return this->_m_time_accumulated_tasking;
}

auto Thread_Pool_ct::time_avg_task_waiting () noexcept -> std::int64_t {
    return this->_m_time_avg_task_waiting;
}

auto Thread_Pool_ct::time_avg_task_executing () noexcept -> std::int64_t {
    return this->_m_time_avg_task_executing;
}

auto Thread_Pool_ct::counts_accumulated_tasks () noexcept -> std::uint64_t {
    return this->_m_counts_accumulated_tasks;
}

auto Thread_Pool_ct::print_state () noexcept -> void {
    std::printf ("Pool State {"
                 "state = %lld, "
                 "max = %lld, "
                 "min = %lld, "
                 "queue = %lld, "
                 "alive = %lld, "
                 "busy = %lld, "
                 "start = %lld, "
                 "stop = %lld, "
                 "accu_runing = %lld, "
                 "accu_tasking = %lld, "
                 "avg_waiting = %lld, "
                 "avg_ema_executing = %lld, "
                 "avg_real_executing = %lld, "
                 "accu_tasks = %lld"
                 "}\n",
                 static_cast<long long int> (this->_m_state.load ()),
                 static_cast<long long int> (this->_m_thread_max.load ()),
                 static_cast<long long int> (this->_m_thread_min.load ()),
                 static_cast<long long int> (this->_m_queue_cur.load ()),
                 static_cast<long long int> (this->_m_thread_alive.load ()),
                 static_cast<long long int> (this->_m_thread_busy.load ()),
                 static_cast<long long int> (this->_m_time_start.load ()),
                 static_cast<long long int> (this->_m_time_stop.load ()),
                 static_cast<long long int> (this->_m_time_accumulated_running.load ()),
                 static_cast<long long int> (this->_m_time_accumulated_tasking.load ()),
                 static_cast<long long int> (this->_m_time_avg_task_waiting.load ()),
                 static_cast<long long int> (this->_m_time_avg_task_executing.load ()),
                 static_cast<long long int> (this->_m_counts_accumulated_tasks ? (this->_m_time_accumulated_tasking / this->_m_counts_accumulated_tasks) : 0),
                 static_cast<long long int> (this->_m_counts_accumulated_tasks.load ()));

    // auto min_thread = Thread_Pool_ct::CountsOF_MinThreads ();
    // auto max_thread = Thread_Pool_ct::CountsOF_MaxThreads (this->_m_time_avg_task_waiting, this->_m_time_avg_task_executing);
    // std::printf ("[suggestion] min thread = %d, max thread = %d\n\n", min_thread, max_thread);
}

auto Thread_Pool_ct::state () noexcept -> Pool_State_et {
    return this->_m_state.load ();
}

auto Thread_Pool_ct::init () noexcept -> bool {
    if (this->_m_state == Pool_State_et::initialized ||
        this->_m_state == Pool_State_et::inRunning ||
        this->_m_state == Pool_State_et::inExiting)
        return false;  // 这些状态下不建议 init

    this->_m_state = Pool_State_et::initialized;

    if (this->_m_thread_min == 0 || this->_m_thread_min <= Thread_Pool_ct::CountsOF_CoreThreads ())
        this->_m_thread_min = Thread_Pool_ct::CountsOF_MinThreads ();

    if (this->_m_thread_max == 0 || this->_m_thread_max <= this->_m_thread_min)
        this->_m_thread_max = this->_m_thread_min + 0;

    if (this->_m_thread_alive != 0)
        this->_m_thread_alive = 0;

    if (this->_m_thread_busy != 0)
        this->_m_thread_busy = 0;

    if (this->_m_thread_step == 0)
        this->_m_thread_step = 1;

    if (this->_m_thread_stepper != 0)
        this->_m_thread_stepper = 0;

    if (this->_m_managing == true)
        this->_m_managing = false;

    if (this->_m_workers.empty () == false)
        this->_m_workers.clear ();

    if (this->_m_queue_max == 0)
        this->_m_queue_max = 200;

    if (this->_m_queue_cur != 0)
        this->_m_queue_cur = 0;

    while (this->_m_tasks.empty () == false)
        this->_m_tasks.pop ();

    return true;
}

auto Thread_Pool_ct::start () noexcept -> bool {
    if (this->_m_state != Pool_State_et::initialized) return false;
    if (this->_m_thread_max == 0 || this->_m_thread_step == 0) return false;

    this->_m_managing = true;
    this->_m_manager  = std::thread {std::bind (&Thread_Pool_ct::_thread_manager, this, nullptr)};
    for (std::uint32_t i = 0; i < this->_m_thread_min; i++)
        this->_m_workers.emplace_back (std::bind (&Thread_Pool_ct::_thread_worker, this, nullptr)),
                this->_m_thread_alive++;

    this->_m_state      = Pool_State_et::inRunning;
    this->_m_time_start = std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();

    return true;
}

auto Thread_Pool_ct::stop () noexcept -> bool {
    if (this->_m_state != Pool_State_et::inRunning)
        return false;

    this->_m_state = Pool_State_et::inExiting;  // 标记为关闭中
    this->_m_not_empty.notify_all ();           // 让工作线程逐渐关闭

    while (this->_m_managing == false);  // 等待管理者线程结束
    this->_m_manager.join ();            // 管理者线程结束

    while (this->_m_thread_alive != 0);                                          // 等待工作线程组结束
    for (auto it = this->_m_workers.begin (); it != this->_m_workers.end ();) {  // "回收" 工作线程
        if (this->_m_exited_ids.count (it->get_id ())) {
            this->_m_exited_ids.erase (it->get_id ());
            it->join ();
            it = this->_m_workers.erase (it);
        } else ++it;
    }

    this->_m_workers.clear ();                                    // 清空工作线程
    this->_m_state                    = Pool_State_et::isExited;  // 标记为已关闭
    this->_m_time_stop                = std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();
    this->_m_time_accumulated_running = this->_m_time_stop - this->_m_time_start;

    return true;
}

auto Thread_Pool_ct::task_add (Task_Item_st && _task, bool _blocking) noexcept -> bool {
    if (this->_m_state != Pool_State_et::inRunning)  // 如果线程池的状态不是正常工作中, 那么拒绝继续添加任务
        return false;

    if (_blocking == false && this->_m_queue_cur == this->_m_queue_max)  // 如果队列满了且不阻塞等待, 则直接返回
        return false;

    std::unique_lock<std::mutex> lock {this->_m_mutex_queue};                                                    // 保护工作队列
    while (_blocking && this->_m_queue_cur == this->_m_queue_max && this->_m_state != Pool_State_et::inExiting)  // 等待任务队列不满
        this->_m_not_full.wait (lock, [this] { return this->_m_tasks.size () < this->_m_queue_max || this->_m_state == Pool_State_et::inExiting; });

    if (this->_m_state == Pool_State_et::inExiting)  // 如果条件变量唤醒时发现线程池处于关闭中了, 则放弃加入工作队列
        return false;

    if (this->_m_queue_cur < this->_m_queue_max) {  // 如果程序能够运行到这里说明这个条件一定是满足的, 这里只是为了逻辑上清晰
        // 该作用域用于真正添加任务
        _task.time_attach () = std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();
        std::time (nullptr);
        _task.state () = Thread_Pool_ct::Task_State_t::newlyBuild;
        this->_m_tasks.push (std::move (_task));
        this->_m_queue_cur++;
        this->_m_not_empty.notify_one ();
    }

    return true;
}

auto Thread_Pool_ct::task_clear () noexcept -> void {
    if (this->_m_state != Pool_State_et::inRunning)
        return;

    if (this->_m_queue_cur == 0)
        return;


    std::unique_lock<std::mutex> lock {this->_m_mutex_queue};
    while (this->_m_queue_cur > 0 || !this->_m_tasks.empty ()) {
        this->_m_tasks.pop ();
        this->_m_queue_cur--;
    }
    this->_m_queue_cur = 0;
    this->_m_not_full.notify_all ();
}

auto Thread_Pool_ct::_thread_manager (void * _args) -> void {
    while (this->_m_state != Pool_State_et::inRunning);  // 等待 start 函数完全结束

    while (this->_m_state == Pool_State_et::inRunning) {                // 管理者线程生命周期为线程池正常工作期间
        std::this_thread::sleep_for (std::chrono::milliseconds (100));  // 任意写的一个延时函数, 每间隔指定长度时间进行一次线程池检查/调整

        // 手动构造一个 unique_lock 的作用域
        {
            std::unique_lock<std::mutex> lock {this->_m_mutex_queue};

            if (this->_m_state == Pool_State_et::inExiting)  // 如果线程池正在关闭, 管理者线程直接退出
                break;

            if (this->_m_auto_threshold) {
                this->_m_thread_min = CountsOF_MinThreads ();
                this->_m_thread_max = CountsOF_MaxThreads (this->_m_time_avg_task_waiting, this->_m_time_avg_task_executing);
            }

            // 这里的标准是自定义的
            // 如果工作队列中积压的未执行的任务的数量大于活动线程的根号2倍并且线程数量没有达到最大值, 则适当增加工作线程数量
            if (this->_m_queue_cur >= this->_m_thread_alive * 1.414 && this->_m_thread_alive < this->_m_thread_max) {

                this->_m_thread_stepper = this->_m_thread_step;  // 设置为正数, 表示需要增加线程数量


                while (this->_m_thread_stepper > 0 && this->_m_thread_alive < this->_m_thread_max) {
                    this->_m_workers.emplace_back (std::bind (&Thread_Pool_ct::_thread_worker, this, nullptr));
                    this->_m_thread_alive++;
                    this->_m_thread_stepper--;
                }
            }

            // 这里的标准也是自定义的
            // 如果正在干活的线程数量的根号2倍还是小于活动线程数量, 说明空闲线程数量过多, 则适当减少工作线程
            if ((this->_m_thread_busy * 1.414) < this->_m_thread_alive && this->_m_thread_alive > this->_m_thread_min) {
                this->_m_thread_stepper = -this->_m_thread_step;  // 设置为负数, 表示要减少线程数量

                for (std::uint32_t i = 0; i < this->_m_thread_step; i++)
                    if (this->_m_thread_alive - i - 1 > this->_m_thread_min)  // 线程数量的减少不能使其低于最低值
                        this->_m_not_empty.notify_one ();
            }

            // 为什么需要清理掉已经结束的线程呢
            // 因为线程的退出是线程自己的行为, 管理者线程无法直接杀死线程, 管理者只能给出指令等待工作线程自己退出
            // 而 std::thread 一旦完成不可以复用, 所以需要定时清理掉那些自发退出的线程
            std::unique_lock<std::mutex> lock_exited_ids {this->_m_mutex_exited_ids};
            for (auto it = this->_m_workers.begin (); it != this->_m_workers.end ();) {
                if (this->_m_exited_ids.count (it->get_id ())) {
                    this->_m_exited_ids.erase (it->get_id ());
                    it->join ();
                    it = this->_m_workers.erase (it);
                } else ++it;
            }
            std::unique_lock<std::mutex> lock_timing {this->_m_mutex_timing};
            this->_m_time_accumulated_running = std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count () - this->_m_time_start;
        }
    }
    this->_m_managing = false;
}

auto Thread_Pool_ct::_thread_worker (void * _args) -> void {

    while (true) {
        {

            std::unique_lock<std::mutex> lock {this->_m_mutex_queue};
            while (this->_m_queue_cur == 0 && this->_m_state != Pool_State_et::inExiting)
                this->_m_not_empty.wait (lock, [this] { return this->_m_tasks.size () > 0 || this->_m_state == Pool_State_et::inExiting; });

            // 如果线程池要关闭了, 则确保队列中的任务已经执行完了, 如果没有则工作线程继续工作直至队列清空
            if (this->_m_state == Pool_State_et::inExiting && this->_m_queue_cur == 0) {
                this->_m_thread_alive--;
                break;
            }

            // 检查线程调节器, 如果小于 0, 表示需要退出一部分线程
            if (this->_m_thread_stepper < 0) {
                this->_m_thread_stepper++;
                this->_m_thread_alive--;
                break;
            }

            Task_Item_st task;
            if (this->_m_queue_cur > 0) {      // 取出任务
                task = this->_m_tasks.top ();  // 按照优先级取出
                this->_m_tasks.pop ();         // 删除队列中的任务
                this->_m_queue_cur--;
                this->_m_not_full.notify_one ();
            }

            this->_m_thread_busy++;  // 标记忙碌中
            lock.unlock ();          // 准备执行任务了 (可能很消耗时间), 所以提前释放锁给其他线程使用
            task.run ();             // 正式执行任务
            // lock.lock ();         // std::unique_lock 在超出作用域析构时会自动检查是否已经 unlock, 所以这条语句是多余的

            if (task.state_isFinished ()) {
                std::unique_lock<std::mutex> lock_timing {this->_m_mutex_timing};
                this->_m_counts_accumulated_tasks++;
                this->_m_time_accumulated_tasking += task.time_cpu ();
                this->_m_time_avg_task_waiting     = Thread_Pool_ct::EMAverage (this->_m_time_avg_task_waiting, task.time_process () - task.time_attach ());
                this->_m_time_avg_task_executing   = Thread_Pool_ct::EMAverage (this->_m_time_avg_task_executing, task.time_cpu ());
            }
            task.clear ();
            this->_m_thread_busy--;
        }
    }
    std::unique_lock<std::mutex> lock_exited_ids {this->_m_mutex_exited_ids};
    this->_m_exited_ids.insert (std::this_thread::get_id ());
}
